package com.haoxin.stream.base

import org.apache.flink.table.api.EnvironmentSettings
import org.apache.flink.table.api.bridge.scala.StreamTableEnvironment

import java.time.ZoneId

trait FlinkTableEnv extends FlinkStreamEnv {

  protected lazy val tableEnv: StreamTableEnvironment = initTabEnv()

  private def initTabEnv() = {
    val settings = EnvironmentSettings.newInstance()
      .useBlinkPlanner()
      .inStreamingMode()
      .build()
    val tEnv = StreamTableEnvironment.create(env, settings)
    tEnv.getConfig.setLocalTimeZone(ZoneId.of("Asia/Shanghai"))
    tEnv
  }

  protected def execute(args: Array[String])

  override protected def run(args: Array[String]): Unit = {
    execute(args: Array[String])
  }
}
